package com.navinfo.opentsp.platform.computing.analysis.application;

import com.lc.core.protocol.common.LCLocationDataOld;
import com.mongodb.spark.MongoSpark;
import com.mongodb.spark.config.ReadConfig;
import com.mongodb.spark.rdd.api.java.JavaMongoRDD;
import com.navinfo.opentsp.platform.computing.analysis.util.PropertiesUtil;
import com.navinfo.opentsp.platform.location.protocol.common.LCFaultInfo;
import com.navinfo.opentsp.platform.location.protocol.common.LCLocationData;
import com.navinfo.opentsp.platform.location.protocol.common.LCVehicleBreakdown;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.SparkSession;
import org.bson.Document;
import org.bson.types.Binary;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import scala.Tuple2;

import java.util.*;

public class MongoToKafka {

    public static void main(String[] args) {
        String day = args[0];
//        String day = "20190409";
        String mongoUri = PropertiesUtil.getProperties("mongoUri");
//        if(args.length >= 2){
//            day = args[0];
//            mongoUri = args[1];
//        }
        SparkSession spark = SparkSession.builder()
                .appName("MongoToKafka")
//                .master("local")
                .config("spark.mongodb.input.uri", mongoUri)
                .config("spark.mongodb.input.partitioner", "MongoPaginateBySizePartitioner")
                .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                .getOrCreate();
        try{
            JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
            exec(day, jsc);
        } catch (Exception e) {
            System.out.println(e.getMessage());
        } finally {
            spark.stop();
        }
    }

    private static void exec(String day, JavaSparkContext jsc) {
        Map<String, String> readOverrides = new HashMap<>();
        String topic = PropertiesUtil.getProperties("topic");
        String topic_fault = PropertiesUtil.getProperties("d001.topic");
        String kafka = PropertiesUtil.getProperties("kafka");
        readOverrides.put("collection", "GpsDataEntity_"+day.substring(2,6));
//        readOverrides.put("readPreference.name", "secondary");
        ReadConfig readConfig = ReadConfig.create(jsc).withOptions(readOverrides);
        Document filter = Document.parse("{ $match: {day : '"+ day + "'} }");
        JavaMongoRDD<Document> rdd = MongoSpark.load(jsc, readConfig).withPipeline(Collections.singletonList(filter));
        System.out.println(rdd.count() + "===================================");
        rdd.mapPartitionsToPair((PairFlatMapFunction<Iterator<Document>, Long, List<Document>>) itor -> {
            List<Tuple2<Long, List<Document>>> ret = new ArrayList<>();
            while(itor.hasNext()){
                Document doc = itor.next();
                Long tid = doc.getLong("tId");
                List<Document> dataList = (List<Document>) doc.get("dataList");
                ret.add(Tuple2.apply(tid, dataList));
            }
            return ret.iterator();
        }).reduceByKey((Function2<List<Document>, List<Document>, List<Document>>) (elements1, elements2) -> {
            elements1.addAll(elements2);
            return elements1;
        }).foreachPartition(new VoidFunction<Iterator<Tuple2<Long, List<Document>>>>() {
            @Override
            public void call(Iterator<Tuple2<Long, List<Document>>> itor) throws Exception {
                Map<String, Object> props = new HashMap<>();
                props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
//        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.30.50.21:9092");
                props.put(ProducerConfig.RETRIES_CONFIG, 0);
                props.put(ProducerConfig.BATCH_SIZE_CONFIG, 100000);
                props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
                props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
                props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
                props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "com.navinfo.opentsp.platform.computing.analysis.entity.ProtoBufSerializer");
                props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);

                KafkaTemplate<String, Object> kafkaTemplate = new KafkaTemplate<String, Object>(new DefaultKafkaProducerFactory<>(props));
                while(itor.hasNext()){
                    Tuple2<Long, List<Document>> tuple2 = itor.next();
                    long tid = tuple2._1;
                    List<Document> dataList = tuple2._2;
                    for (Document data : dataList) {
                        Binary binary = (Binary) data.get("data");
                        LCLocationDataOld.LocationData locationDataOld = null;
                        LCLocationData.LocationData locationData = null;
                        LCFaultInfo.FaultInfo.Builder builder = LCFaultInfo.FaultInfo.newBuilder();
                        LCFaultInfo.GpsLocationData.Builder gpsLocationData = LCFaultInfo.GpsLocationData.newBuilder();
                        LCFaultInfo.FaultListData.Builder faultListData = LCFaultInfo.FaultListData.newBuilder();



                        try {
                            locationDataOld = LCLocationDataOld.LocationData.parseFrom(binary.getData());
                            locationData = LCLocationData.LocationData.parseFrom(binary.getData());
                            LCLocationData.LocationData newLocationData = locationData.toBuilder()
                                    .setTerminalId(tid)
                                    .setStayTime(0)
                                    .setSerialNumber(0)
                                    .setGearStatus(locationDataOld.getGearStatus())
                                    .setTcuLoad(locationDataOld.getTcuLoad())
                                    .setVecuLoad(locationDataOld.getVecuLoad())
                                    .setUreaInjectionAvg(locationDataOld.getUreaInjectionAvg())
                                    .setUreaPressureAvg(locationDataOld.getUreaPressureAvg()).build();
                            kafkaTemplate.send(topic, tid + "", newLocationData.toByteArray());

                           if (newLocationData.getBreakdownAddition().getBreakdownCount()>0){
                               gpsLocationData.setAlarm(newLocationData.getAlarm());
                               gpsLocationData.setLatitude(newLocationData.getLatitude());
                               gpsLocationData.setDirection(newLocationData.getDirection());
                               gpsLocationData.setGpsTime(newLocationData.getGpsDate());
                               gpsLocationData.setHeight(newLocationData.getHeight());
                               gpsLocationData.setLongitude(newLocationData.getLongitude());
                               gpsLocationData.setSpeed(newLocationData.getSpeed());
                               gpsLocationData.setStatus(newLocationData.getStatus());

                               builder.setGpsLocationData(gpsLocationData);
                               builder.setMark(0);
                               builder.setTerminalId(newLocationData.getTerminalId());
                               builder.setOriginalLat(newLocationData.getOriginalLat());
                               builder.setOriginalLng(newLocationData.getOriginalLng());

                               List<LCVehicleBreakdown.VehicleBreakdown> faultList = newLocationData.getBreakdownAddition().getBreakdownList();
                               for (Iterator<LCVehicleBreakdown.VehicleBreakdown> iterator = faultList.iterator(); iterator.hasNext(); ) {
                                   LCVehicleBreakdown.VehicleBreakdown next =  iterator.next();
                                   LCFaultInfo.FaultItem.Builder faultItem = LCFaultInfo.FaultItem.newBuilder();;
                                   // 20200206 东风李勇确认,故障原地址是49（31），
                                   // 并且SPN为0、65536（10000H）、131072（20000H）、
                                   // 196608（30000H）、262144（40000H）、327680（50000H）、
                                   // 393216（60000H）、458752（70000H）的故障码，
                                   // 不用判断FMI，可直接判断为异常故障，直接舍弃
                                   if (!(next.getFaultType() == 49 &&(next.getBreakdownSPNValue()!=0 || next.getBreakdownSPNValue()!=65536 || next.getBreakdownSPNValue()!=131072
                                           || next.getBreakdownSPNValue()!= 196608 || next.getBreakdownSPNValue()!= 262144
                                           || next.getBreakdownSPNValue()!= 327680 || next.getBreakdownSPNValue()!= 393216))) {
                                       faultItem.setSPN(next.getBreakdownSPNValue());
                                       faultItem.setFMI(next.getBreakdownFMIValue());
                                       faultItem.setFaultAddr(next.getFaultType());
                                       faultListData.addFaultItem(faultItem);
                                   }
                               }
                               builder.addFaultListData(faultListData);

                               kafkaTemplate.send(topic_fault, tid + "", builder.build().toByteArray());

                           }
                        } catch (Exception e) {
                            //                        logger.error("", e);
                            continue;
                        }
//                    Packet outPacket = new Packet();
//                    outPacket.setCommand(12290);
//                    outPacket.setProtocol(2);
//                    outPacket.setUniqueMark(tid + "");
//                    outPacket.setSerialNumber(1111);
//                    outPacket.setContent(newLocationData.toByteArray());

//                    KafkaCommand kafkaCommand = new KafkaCommand();
//                    kafkaCommand.setKey(tid + "");
//                    kafkaCommand.setCommandId("3002");
//                    kafkaCommand.setSendTime(new Date());
//                    kafkaCommand.setSerialNumber(11111);
//                    kafkaCommand.setTopic("0200spark");
//                    kafkaCommand.setMessage(JsonUtil.pojo2Bytes(outPacket));
//                    kafkaTemplate.send("0200spark", tid + "", JsonUtil.pojo2Json(kafkaCommand));
                    }
                }
            }
        });
    }

    public static Map<String, Object> producerConfigs(String kafka) {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
//        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.30.50.21:9092");
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 100000);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "com.navinfo.opentsp.platform.computing.analysis.entity.ProtoBufSerializer");
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        return props;
    }
}
